/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.ratings.normilize.defaults.reducers;

import aiaudio.processing.base.mapreduce.CoordinateDoubleValueKey;
import aiaudio.Application;
import aiaudio.database.tablegroup.InitialDataTableGroup;
import aiaudio.database.tables.RatingsMatrixTable;
import aiaudio.database.tables.UsersListTable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;

/**
 *
 * @author nastya
 */
public abstract class SimpleRatingUserToArtistValueReducer extends TableReducer<ImmutableBytesWritable, CoordinateDoubleValueKey, ImmutableBytesWritable> {

    @Override
    protected void reduce(ImmutableBytesWritable key, Iterable<CoordinateDoubleValueKey> values, Context context) throws IOException, InterruptedException {
        long totalVotingByUser = getTotalVotingByUser(key.get());

        List<CoordinateDoubleValueKey> lst = new ArrayList<CoordinateDoubleValueKey>();

        double sum = 0;
        double max = Double.MIN_VALUE;
        double min = Double.MAX_VALUE;
        for (CoordinateDoubleValueKey artistValueKey : values) {
            lst.add(new CoordinateDoubleValueKey(artistValueKey.getValue(), artistValueKey.getCoordinate()));

            double value = artistValueKey.getValue();
            if (value > max) {
                max = value;
            }
            if (value < min) {
                min = value;
            }

            sum += value;
        }

        double sub = max - min;

        HashMap<String, Double> params = new HashMap<String, Double>();
        params.put("sum", sum);
        params.put("min", min);
        params.put("max", max);
        params.put("total", (double) totalVotingByUser);
        double defaultVoting = calculateDefaultVoting(params);

        Put put = new Put(key.get());

        for (CoordinateDoubleValueKey artistValueKey : lst) {
            double result = 1;
            if (sub != 0) {
                result = (artistValueKey.getValue() - min) / sub;
            }

            RatingsMatrixTable.addUserRating(put, artistValueKey.getCoordinate(), result);
        }

        double defaultVotingProcessed = sub == 0 ? 1 : (defaultVoting - min) / sub;
        RatingsMatrixTable.addDefaultRating(put, defaultVotingProcessed);

        context.write(key, put);
    }

    private long getTotalVotingByUser(byte[] user) throws IOException {
        HTable usersTable = Application.database().getByName(InitialDataTableGroup.getInstance().getUsersTable()).getTable();
        return UsersListTable.getVotingCount(usersTable, user);
    }

    protected abstract double calculateDefaultVoting(HashMap<String, Double> calculatedParams);
}
